package com.bw.util;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Set;


public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    public static List<String> cloumnList = ListUtil.toList("key", "value", "values", "like", "table", "view", "date", "datetime");
    private Connection connection;

    private String schema; 
    private String info;

    public DimSinkFunction() {

    }

    public DimSinkFunction(String schema) {
        this.schema = schema;
    }

    public DimSinkFunction(String schema, String info) {
        this.schema = schema;
        this.info = info;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_URL);
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        PreparedStatement preparedStatement = null;
        try {
            String sinkTable = value.getJSONObject("source").getString("table");
            sinkTable = "dim_" + sinkTable;
            JSONObject data = value.getJSONObject("after");
            Set<String> strings = Sets.newHashSet(data.keySet());
            strings.remove("id");
            String sinkColumns = StringUtils.join(strings, ",");
            buildSchema();
            buildTable(sinkTable, sinkColumns, null, null);
            String upsertSql = genUpsertSql(sinkTable,
                    data);
            System.out.println(upsertSql);
            preparedStatement = connection.prepareStatement(upsertSql);
            preparedStatement.executeUpdate();

        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }

    }

    private void buildSchema() {
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement("create schema if not exists " + schema);
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("Phoenix database " + schema + " create fail！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private String genUpsertSql(String sinkTable, JSONObject data) {
        Object id = data.get("id");
        data.remove("id");
        JSONObject newData = new JSONObject();
        data.forEach((k, v) -> {
            if (v == null) {
                v = "";
            }
            if (StrUtil.isBlank(info)) {
                newData.put(wrapperField(k), v.toString().replace("'", ""));
            } else {
                newData.put(info + "." + wrapperField(k), v.toString().replace("'", ""));
            }
        });

        return "upsert into " + schema + "." + sinkTable + "(id," +
                StringUtils.join(newData.keySet(), ",") + ") values(" + id + ", '" +
                StringUtils.join(newData.values(), "','") + "')";
    }

    private void buildTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {

        PreparedStatement preparedStatement = null;

        try {
            if (sinkPk == null) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }

            StringBuffer createTableSQL = new StringBuffer("create table if not exists ")
                    .append(schema)
                    .append(".")
                    .append(sinkTable)
                    .append("( id bigint primary key ,");

            String[] fields = sinkColumns.split(",");
            for (int i = 0; i < fields.length; i++) {
                String field;
                if (StrUtil.isBlank(info)) {
                    field = wrapperField(fields[i]);
                } else {
                    field = info + "." + wrapperField(fields[i]);
                }

                createTableSQL.append(field).append(" varchar ");

                if (i < fields.length - 1) {
                    createTableSQL.append(",");
                }
            }

            createTableSQL.append(")").append(sinkExtend);

            System.out.println(createTableSQL);

            preparedStatement = connection.prepareStatement(createTableSQL.toString());
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("Phoenix table" + sinkTable + " create fail ！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    String wrapperField(String o) {

        if (cloumnList.contains(o.toLowerCase())) {
            o += "_v2";
        }
        return o;
//        return CaseFormat.LOWER_UNDERSCORE
//                .to(CaseFormat.LOWER_CAMEL, o.toLowerCase());
    }
}
